package com.zhike.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.TripBean;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author: zhike
 * @CreateTime: 2024/3/11 9:37
 * @Description:
 */
public class testReduce {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        //2.读取car_electricity_log的数据并创建流
        DataStreamSource<String> electricityDS = env.fromSource(KafkaUtils.getKafkaSource("car_electricity_log", "dws_trip_window_1"), WatermarkStrategy.noWatermarks(),"ele_source");

        //转换数据类型
        SingleOutputStreamOperator<JSONObject> eleJsonDS = electricityDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSON.parseObject(value);
            }
        });

        SingleOutputStreamOperator<TripBean> eleTripDS = eleJsonDS.map(new MapFunction<JSONObject, TripBean>() {

            @Override
            public TripBean map(JSONObject value) throws Exception {
                return getFields(value);
            }
        });
        eleTripDS.print();

        eleTripDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TripBean>forMonotonousTimestamps()
                        .withTimestampAssigner((element, timestamp) -> element.getTs())
        ).windowAll(TumblingEventTimeWindows.of(Time.seconds(30L)))
                .reduce(new ReduceFunction<TripBean>() {
                    @Override
                    public TripBean reduce(TripBean value1, TripBean value2) throws Exception {
                        value1.setMinMileage(value1.getTotalSpeed() + value2.getTotalSpeed());
                        System.out.println("来数据啦！");
                        return value1;
                    }
                }, new AllWindowFunction<TripBean, TripBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TripBean> values, Collector<TripBean> out) throws Exception {
                        TripBean tripBean = values.iterator().next();
                        tripBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        tripBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        tripBean.setTs(System.currentTimeMillis());
                        tripBean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        out.collect(tripBean);
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    private static TripBean getFields(JSONObject value) {
        //获取汽车id
        String vin = value.getString("vin");
        //获取里程
        double mileage = value.getInteger("mileage") / 10.0;
        //获取当前速度（分子）
        double velocity = value.getInteger("velocity") / 10.0;
        //获取时间戳
        Long ts = value.getLong("timestamp");
        return TripBean.builder()
                .vin(vin)
                .maxMileage(mileage)
                .totalSpeed(velocity)
                .ts(ts)
                .build();
    }
}
